package com.origin.src.mq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class RocketMQ_XA {

    static String nameServer = "127.0.0.1:9876;127.0.0.1:9877";
    static String topicName = "TopicTX";
    static String tag_XA_Order = "tag_XA_Order";
    static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            1, 8, 3000L, TimeUnit.SECONDS
            , new LinkedBlockingQueue<>(4));




    static AtomicInteger saveI = new AtomicInteger(0);

    public static String saveOrder(Object order) {
        int i = saveI.incrementAndGet();
        System.out.println(ConsoleColorFont.msg(Thread.currentThread().getName() + " saveOrder:" + i, order.toString()));
        return order.toString();
    }

    static AtomicInteger checkI = new AtomicInteger(0);

    public static String checkOrder(Object order) {
        int i = checkI.incrementAndGet();
        System.out.println(ConsoleColorFont.msg(Thread.currentThread().getName() + " checkOrder:" + i, order.toString()));
        return i < 15 ? order.toString() : "1";
    }

    public static void producer() throws MQClientException, InterruptedException {
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("tx_producer");
        transactionMQProducer.setNamesrvAddr(nameServer);

        transactionMQProducer.setExecutorService(threadPoolExecutor);
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                switch (saveOrder(o)) {
                    case "1":
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case "0":
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    default:
                        return LocalTransactionState.UNKNOW;
                }

            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                switch (checkOrder(msg)) {
                    case "1":
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case "0":
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    default:
                        return LocalTransactionState.UNKNOW;
                }
            }


        });

        transactionMQProducer.start();


        Message message = new Message(topicName, tag_XA_Order, "2".getBytes(StandardCharsets.UTF_8));

        TransactionSendResult transactionSendResult = transactionMQProducer.sendMessageInTransaction(message, "2");


        while (transactionSendResult.getLocalTransactionState().equals(LocalTransactionState.UNKNOW)) {
            System.out.println(transactionSendResult.getLocalTransactionState());
            Thread.sleep(15000);
        }
        System.out.println("shutdown");
        transactionMQProducer.shutdown();
    }

    public static void main(String[] args) throws InterruptedException, MQClientException {
        producer();
    }



}
